package store

import (
	"fmt"
	"sync"

	"github.com/go-redis/redis"
)

type EdgeChannelStore interface {
	Set(m EdgeChannelData) error
	Get(id string) (*EdgeChannelData, error)
	Subscribe(pattern string) (Subscription, error)
}

type StoreSettings struct {
	Addr      string
	Password  string
	DB        int
	KeyPrefix string
}

var once = sync.Once{}

type store struct {
	settings StoreSettings
	client   *redis.Client
}

var gs *store

func New(settings StoreSettings) (EdgeChannelStore, error) {
	var err error
	once.Do(func() {
		gs, err = newStore(settings)
	})
	return gs, err
}

func newStore(settings StoreSettings) (*store, error) {
	options := redis.Options{
		Addr:     settings.Addr,
		DB:       settings.DB,
		Password: settings.Password,
	}
	var client *redis.Client
	client, err := initRedis(options)
	if err != nil {
		return nil, err
	}
	s := &store{
		settings: settings,
		client:   client,
	}
	return s, nil
}

func initRedis(settings redis.Options) (*redis.Client, error) {
	client := redis.NewClient(&settings)
	return client, nil
}

func (s *store) Set(m EdgeChannelData) error {
	key := s.formatKey(m.ID)
	value := serializeMessage(m)
	tx := s.client.TxPipeline()
	tx.Set(key, value, 0)
	tx.Publish(key, value)
	_, err := tx.Exec()
	return err
}

func (s *store) Get(id string) (*EdgeChannelData, error) {
	key := s.formatKey(id)
	r, err := s.client.Get(key).Result()
	if err == redis.Nil {
		return nil, nil
	} else if err != nil {
		return nil, err
	} else {
		m, err := ParseEdgeChannelData(r)
		if err != nil {
			return nil, err
		}
		return &m, nil
	}
}

type Subscription interface {
	Channel() chan EdgeChannelData
	Unsubscribe()
}

type subscription struct {
	channel chan EdgeChannelData
	exit    chan int
}

func (sub *subscription) Channel() chan EdgeChannelData {
	return sub.channel
}

func (sub *subscription) Unsubscribe() {
	sub.exit <- 1
}

func (s *store) Subscribe(pattern string) (Subscription, error) {
	key := s.formatKey(pattern)
	pkey := s.formatPattern(key)
	rsub := s.client.PSubscribe(pkey)
	channel := make(chan EdgeChannelData)
	exit := make(chan int)
	go func() {
		for {
			select {
			case m := <-rsub.Channel():
				{
					md, err := parseMessage(m)
					if err == nil {
						channel <- md
					}
				}
			case <-exit:
				{
					return
				}
			}
		}
	}()
	sub := subscription{
		channel: channel,
		exit:    exit,
	}
	return &sub, nil
}

func (s *store) formatKey(id string) string {
	if s.settings.KeyPrefix == "" {
		return id
	}
	return fmt.Sprintf("%s/%s", s.settings.KeyPrefix, id)
}

func (s *store) formatPattern(key string) string {
	return key
}

func serializeMessage(m EdgeChannelData) string {
	return m.Serialize()
}

func parseMessage(m *redis.Message) (EdgeChannelData, error) {
	return ParseEdgeChannelData(m.Payload)
}
